package stc.skymobi.netty.filter;

import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
import stc.skymobi.bean.esb.core.AbstractEsbA2MSignal;
import stc.skymobi.bean.esb.core.TerminalAccessInfo;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 校验A2M的请求中,是否含有合法的skyId
 */
@ManagedResource
public class A2MSignalSkyIdFilter implements Filter {
    private static final Logger logger = LoggerFactory.getLogger(A2MSignalSkyIdFilter.class);
    private final Map<Long, SkyIdCountVo> userRequestCount = new ConcurrentHashMap<Long, SkyIdCountVo>();
    private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
    private boolean enableCounter;
    private int countPeriod = 60;

    @Override
    public boolean doFilter(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof AbstractEsbA2MSignal) {
            long skyId = -1;
            AbstractEsbA2MSignal signal = (AbstractEsbA2MSignal) msg;
            TerminalAccessInfo terminalAccessInfo = signal.getTerminalAccessInfo();
            if (terminalAccessInfo != null) {
                skyId = terminalAccessInfo.getSkyId();
            }

            if (enableCounter) {
                SkyIdCountVo count = userRequestCount.get(skyId);
                if (count == null) {
                    count = new SkyIdCountVo();
                    count.setSkyId(skyId);
                    userRequestCount.put(skyId, count);
                }
                count.setCount(count.getCount() + 1);
                if (logger.isDebugEnabled()) {
                    logger.debug("{} send an a2m request {}", skyId, msg);
                }
            }

            if (skyId <= 0) {
                logger.warn("skyId {} is invalid,signal {}", skyId, signal);
                return false;
            }

            return true;
        }
        return true;
    }

    @ManagedAttribute(description = "是否启用用户请求计数")
    public void setEnableCounter(boolean enableCounter) {
        this.enableCounter = enableCounter;
    }

    @ManagedAttribute
    public boolean isEnableCounter() {
        return enableCounter;
    }

    @ManagedAttribute(description = "用户请求计数详情,查看时对系统性能有影响,不要经常查看")
    public String getUserRequestCountDetails() {
        List<SkyIdCountVo> counts = new ArrayList<SkyIdCountVo>(userRequestCount.values());
        Collections.sort(counts);
        return counts.toString();
    }

    @ManagedAttribute(description = "用户请求计数Map大小")
    public int getUserRequestCountSize() {
        return userRequestCount.size();
    }

    @ManagedAttribute(description = "用户请求计数周期,单位秒,最大值10分钟")
    public void setCountPeriod(int countPeriod) {
        if (countPeriod <= 600) {
            this.countPeriod = countPeriod;
        } else {
            logger.warn("countPeriod can't larger than 600 seconds");
        }
    }

    @ManagedAttribute
    public int getCountPeriod() {
        return countPeriod;
    }

    @PostConstruct
    public void init() throws Exception {
        executorService.schedule(new Runnable() {
            @Override
            public void run() {
                userRequestCount.clear();
                executorService.schedule(this, countPeriod, TimeUnit.SECONDS);
            }
        }, countPeriod, TimeUnit.SECONDS);
    }
}

class SkyIdCountVo implements Comparable<SkyIdCountVo> {
    private long skyId;
    private Integer count = 0;

    public long getSkyId() {
        return skyId;
    }

    public void setSkyId(long skyId) {
        this.skyId = skyId;
    }

    public Integer getCount() {
        return count;
    }

    public void setCount(int count) {
        this.count = count;
    }

    @Override
    public int compareTo(SkyIdCountVo o) {
        return o.getCount().compareTo(this.getCount());
    }

    @Override
    public String toString() {
        return skyId + ":" + count;
    }
}
